package com.office.mqtt;

import cn.hutool.json.JSONUtil;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.office.entity.OfficeData;
import com.office.mapper.OfficeDataMapper;
import com.office.webSocket.WebSocket;
import org.apache.logging.log4j.util.Strings;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.core.MessageProducer;
import org.springframework.integration.mqtt.core.DefaultMqttPahoClientFactory;
import org.springframework.integration.mqtt.core.MqttPahoClientFactory;
import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter;
import org.springframework.integration.mqtt.outbound.MqttPahoMessageHandler;
import org.springframework.integration.mqtt.support.DefaultPahoMessageConverter;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;

import javax.annotation.Resource;
import java.text.SimpleDateFormat;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.Date;
import java.util.Objects;
import java.util.logging.SimpleFormatter;

@Configuration
public class MqttConfig {

    @Resource
    private WebSocket webSocket;
    @Resource
    private OfficeDataMapper officeDataMapper;

    @Bean
    public MqttPahoClientFactory mqttClientFactory() {
        DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();
        MqttConnectOptions options = new MqttConnectOptions();

        //设置MQTT代理端URL和端口，可以设置多个
        options.setServerURIs(new String[]{"tcp://test.mosquitto.org:1883"});
        //options.setServerURIs(new String[]{"tcp://test.mosquitto.org:1883"});

        factory.setConnectionOptions(options);
        return factory;
    }

    @Bean
    public MessageChannel mqttInputChannel() {
        return new DirectChannel();
    }

    @Bean
    public MessageProducer inbound() {
        //获取时间戳作为clientId，防止连接一段时间后自动断开
        String clientId = new SimpleDateFormat("yyyyMMddHHmmss").format(new Date());
        //设置clientId和需要订阅的主题，可以设置多个
        MqttPahoMessageDrivenChannelAdapter adapter = new MqttPahoMessageDrivenChannelAdapter(clientId,
                mqttClientFactory(), "office/device", "off", "123");

        adapter.setCompletionTimeout(5000);
        DefaultPahoMessageConverter defaultPahoMessageConverter = new DefaultPahoMessageConverter();
        adapter.setConverter(defaultPahoMessageConverter);
        adapter.setQos(1);
        adapter.setOutputChannel(mqttInputChannel());
        return adapter;
    }

    @Bean
    @ServiceActivator(inputChannel = "mqttInputChannel")
    public MessageHandler handler() {
        return message -> {
            //收到MQTT的入栈消息，可以根据自己在发布端发布的消息内容，进行不同处理
            String payload = message.getPayload().toString();
            //筛选MQTT主题
            String topic = Objects.requireNonNull(message.getHeaders().get("mqtt_receivedTopic")).toString();

            // 根据主题消息处理。
            if (topic.equals("office/device")) {
                System.out.println("收到办公室采集器消息： " + payload);

                //  10/10/10/10/10
                //  temp/humi/light/smog/sound
                String[] split = payload.split("/");
                String temp = String.valueOf(Double.valueOf(split[1]).intValue());
                String humi = String.valueOf(Double.valueOf(split[2]).intValue());
                String light = split[3];
                String smog = split[4];
                String sound = split[5];

                JSONObject payloadJson = new JSONObject();
                payloadJson.put("temp", temp);
                payloadJson.put("humi", humi);
                payloadJson.put("light", light);
                payloadJson.put("smog", smog);
                payloadJson.put("sound", sound);

                //利用webSocket实时发送数据到前端
                if (checkData(temp, 100, 0) &&
                        checkData(humi, 150, 0) &&
                        checkData(smog, 1000, 0) &&
                        checkData(light, 1100, 0) &&
                        checkData(sound, 1100, 0)) {
                    webSocket.sendOfficeData(payloadJson);
                    System.out.println(payloadJson);
                    System.out.println("办公室采集器消息已经发送到前端");
                }

                //将MQTT消息数据保存到数据库
                OfficeData officeData = new OfficeData();

                if (!Strings.isBlank(temp) && temp.length() <= 3) {
                    officeData.setTemp(temp);
                }
                if (!Strings.isBlank(humi) && humi.length() <= 3) {
                    officeData.setHumi(humi);
                }
                if (!Strings.isBlank(smog) && smog.length() <= 3) {
                    officeData.setSmog(smog);
                }
                if (!Strings.isBlank(light) && light.length() <= 4) {
                    officeData.setLight(light);
                }
                if (!Strings.isBlank(sound) && sound.length() <= 4) {
                    officeData.setSound(sound);
                }
                if (Strings.isNotBlank(officeData.toString())) {
                    officeData.setTime(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss").format(LocalDateTime.now()));
                    officeDataMapper.insert(officeData);
                    System.out.println(officeData);
                    System.out.println("办公室采集器消息已经存入数据库");
                }
            }

        };
    }

    @Bean
    public MessageChannel mqttOutboundChannel() {
        return new DirectChannel();
    }

    @Bean
    @ServiceActivator(inputChannel = "mqttOutboundChannel")
    public MessageHandler outbound() {
        //获取时间戳作为clientId，防止连接一段时间后自动断开
        String clientId = new SimpleDateFormat("yyyyMMddHHmmss").format(new Date());
        MqttPahoMessageHandler messageHandler = new MqttPahoMessageHandler(clientId, mqttClientFactory());
        messageHandler.setAsync(true);
        messageHandler.setDefaultTopic("command");
        messageHandler.setDefaultQos(1); // 设置默认QoS
        DefaultPahoMessageConverter defaultPahoMessageConverter = new DefaultPahoMessageConverter();
        messageHandler.setConverter(defaultPahoMessageConverter);
        return messageHandler;
    }

    /**
     * checkData
     * 数据大小校验
     *
     * @param data
     * @param max
     * @param min
     * @return boolean
     **/
    public boolean checkData(String data, Integer max, Integer min) {
        return !Strings.isBlank(data) && Integer.parseInt(data) >= min && Integer.parseInt(data) <= max;
    }

}
